package com.sduept.component.message.core;

import com.sduept.nwld.dataserver.manager.etl.RequestUnfinishTaskETLManager;
import lombok.extern.slf4j.Slf4j;
import net.sf.json.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

import java.util.Optional;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Slf4j
@Service
public class RequestUnfinishTaskService {

    @Autowired
    private RequestUnfinishTaskETLManager taskManager;


    private ExecutorService executor = Executors.newFixedThreadPool(10);

    @KafkaListener(topics = {"request_unfinish_task"})
    public void onMessage(ConsumerRecord<?, ?> record) {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            JSONObject jo = JSONObject.fromObject(kafkaMessage.get());
            final String msgContent = jo.getString("content");
            log.info("接收到消息：request_unfinish_task, content：" + msgContent);
            if (StringUtils.isNotEmpty(msgContent)) {
                executor.submit(new Runnable() {
                    @Override
                    public void run() {
                        //数据源告警消息解析
                        taskManager.receiveAndSaveJsonData(msgContent);
                    }
                });
            }
        }
    }
}
